package com.qiaofang.housewarehouse.dao.hbase;

import com.qiaofang.housewarehouse.dao.utils.ConfigUtils;
import com.qiaofang.housewarehouse.stub.constants.TableConstants;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

public class HbaseClient {

    private static final Logger logger = LoggerFactory.getLogger(HbaseClient.class);
    public static Admin admin = null;
    public static Configuration config = null;
    public static Connection conn = null;

    static {
        try {
            logger.info("hbase初始化配置开始");
            config = HBaseConfiguration.create();
            config.set("hbase.zookeeper.quorum", ConfigUtils.getHbaseZookeeperQuorum());
            config.set("hbase.zookeeper.property.clientPort", ConfigUtils.getHbasePropertyClientPort());
            System.setProperty("HADOOP_USER_NAME", "hdfs");
            conn = ConnectionFactory.createConnection(config);
            admin = conn.getAdmin();
            logger.info("hbase初始化配置====SUCCESS");
        } catch (Exception e) {
            logger.error("hbase初始化配置失败:{}",e.getMessage(),e);
        }
    }

    /**
     * 插入一条记录
     * @param tableName 表名
     * @param rowKey kowkey
     * @param datas 记录
     * @throws IOException
     */
    public static void insertRow(String tableName, String rowKey, List<HbaseColumn> datas) throws IOException {
        Table table = conn.getTable(TableName.valueOf(tableName));
        Put putRow = new Put(Bytes.toBytes(rowKey));
        for (HbaseColumn col : datas) {
            putRow.addColumn(Bytes.toBytes(col.getFamilyName()), Bytes.toBytes(col.getColName()), Bytes.toBytes(col.getColValue()));
        }
        table.put(putRow);
    }

    public static List<Result> getResultListScann(String tableName, String start_rowkey, String stop_rowkey) throws IOException {
        Table table = null;
        ResultScanner rs = null;
        List<Result> datas = new ArrayList<>();
        try {
            table = conn.getTable(TableName.valueOf(tableName));
            Scan scan = new Scan();
            scan.setStartRow(Bytes.toBytes(start_rowkey));
            scan.setStopRow(Bytes.toBytes(stop_rowkey));
            rs = table.getScanner(scan);
            for (Result r : rs) {
                datas.add(r);
            }
        } finally {
            tableAndRS(table, rs);
        }
        return datas;
    }


    /**
     * 遍历查询hbase表(从 start_rowkey 到 stop_rowkey )
     *
     * @param tableName    表名
     * @param start_rowkey 开始rowkey
     * @param stop_rowkey  结束rowkey
     * @return listHbaseRow
     * @throws IOException
     */
    public static List<HBaseRow> getResultScann(String tableName, String start_rowkey, String stop_rowkey) throws IOException {
        Table table = null;
        ResultScanner rs = null;
        List<HBaseRow> datas = new ArrayList<>();
        try {
            table = conn.getTable(TableName.valueOf(tableName));
            Scan scan = new Scan();
            scan.setStartRow(Bytes.toBytes(start_rowkey));
            scan.setStopRow(Bytes.toBytes(stop_rowkey));
            rs = table.getScanner(scan);
            for (Result r : rs) {
                HBaseRow row = new HBaseRow();
                row.setRowKey(Bytes.toString(r.getRow()));
                List<HbaseColumn> dds = new ArrayList<>();
                for (Cell cell : r.rawCells()) {
                    row.setRowKey(new String(CellUtil.cloneRow(cell)));
                    dds.add(new HbaseColumn(new String(CellUtil.cloneFamily(cell)), new String(CellUtil.cloneQualifier(cell)), new String(CellUtil.cloneValue(cell))));
                }
                row.setDatas(dds);
                datas.add(row);
            }
        } finally {
            tableAndRS(table, rs);
        }
        return datas;
    }

    /**
     * 获取多个rowkey记录数
     *
     * @param tableName
     * @param rowKeys
     * @return
     * @throws Exception
     */
    public static List<Result> getRowKeys(String tableName, List<String> rowKeys) throws Exception {
        Table table = null;
        List<Get> gets = new ArrayList<>();
        try {
            table = conn.getTable(TableName.valueOf(tableName));
            for (String rk : rowKeys) {
                Get get = new Get(Bytes.toBytes(rk));
                gets.add(get);
            }
            return Arrays.asList(table.get(gets));
        } finally {
            try {
                if (null != table) {
                    table.close();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 获取单个rowkey记录数
     *
     * @param tableName
     * @param rowKey
     * @return
     * @throws Exception
     */
    public static Result getRowKey(String tableName, String rowKey) throws Exception {
        Table table = null;
        try {
            table = conn.getTable(TableName.valueOf(tableName));
            Get get = new Get(Bytes.toBytes(rowKey));
            Result result = table.get(get);
            if(null == result.listCells()){
                logger.info("在{}中没有rowkey={}记录",tableName, rowKey);
            }
            return result;
        } finally {
            try {
                if (null != table) {
                    table.close();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    public static List<Result> getResultsByFilter(String tableName, FilterList filterList ) throws Exception {
        Table table = null;
        ResultScanner rs = null;
        List<Result> result = new ArrayList<>();
        try{
            table = conn.getTable(TableName.valueOf(tableName));
            Scan scan = new Scan();
            scan.setFilter(filterList);
            rs = table.getScanner(scan);
            for (Result res : rs) {
                result.add(res);
            }
        } finally {
            tableAndRS(table, rs);
        }
        return result;
    }

    private static void tableAndRS(Table table, ResultScanner rs) {
        try {
            if (null != table) {
                table.close();
            }
            if (null != rs) {
                rs.close();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) throws Exception {
        HbaseClient.getRowKey(TableConstants.HSWH_INFO_INTEGRITY_COUNTERS, "住宅个人房源出租20170301");
    }

}
